package com.afdor.rws.event.driven.channel.local.observer;

import com.afdor.rws.event.driven.commons.ThreadUtils;
import com.afdor.rws.event.driven.entities.Event;
import com.afdor.rws.event.driven.entities.holder.ResultEvent;
import lombok.EqualsAndHashCode;
import lombok.Getter;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.ConcurrentMap;

/**
 * 事件监听器管理容器，该类中，事件堆积及监听器等操作<font color="red">均为多线程安全</font>的，故可放心使用
 *
 * @author 悭梵
 * @date Created in 2018-08-08 15:12
 */
@Slf4j
public class ObserverEventSource {
    /**
     * 事件队列
     */
    @Getter
    private final ConcurrentMap<String, Queue<ObserverEvent>> eventSourceQueueMap;
    /**
     * 事件监听器列表
     */
    @Getter
    private final ConcurrentMap<String, EventSourceItem> consumerListenerMap;

    public ObserverEventSource() {
        this.consumerListenerMap = new ConcurrentHashMap<String, EventSourceItem>();
        this.eventSourceQueueMap = new ConcurrentHashMap<String, Queue<ObserverEvent>>();
    }

    /**
     * 给事件源注册监听器
     *
     * @param listener
     */
    public synchronized void addEventListener(@NonNull String eventSourceName, @NonNull String consumerName, @NonNull ObserverEventListener listener) {
        EventSourceItem eventSourceItem = this.consumerListenerMap.get(eventSourceName);
        if (eventSourceItem == null) {
            this.consumerListenerMap.putIfAbsent(eventSourceName, new EventSourceItem(eventSourceName));
            eventSourceItem = this.consumerListenerMap.get(eventSourceName);
        }
        ConsumerItem consumerItem = eventSourceItem.getConsumer(consumerName);
        if (consumerItem == null) {
            eventSourceItem.addConsumer(new ConsumerItem(eventSourceName, consumerName));
            consumerItem = eventSourceItem.getConsumer(consumerName);
        }
        consumerItem.addEventListener(listener);
        // 复制历史消息到sub队列
        Queue<ObserverEvent> historyQueue = getEventSourceQueueMap().get(eventSourceName);
        if (!(historyQueue == null || historyQueue.isEmpty())) {
            consumerItem.getEventQueue().addAll(historyQueue);
        }
    }

    /**
     * 清空监听器
     */
    public synchronized void clearEventListener() {
        this.consumerListenerMap.clear();
    }

    /**
     * 清空监听器
     */
    public synchronized void clearEventListener(@NonNull String eventSourceName) {
        this.consumerListenerMap.remove(eventSourceName);
    }

    /**
     * 清空监听器
     */
    public synchronized void clearEventListener(@NonNull String eventSourceName, @NonNull String consumerName) {
        EventSourceItem eventSourceItem = this.consumerListenerMap.get(eventSourceName);
        if (eventSourceItem != null) {
            eventSourceItem.removeConsumer(consumerName);
        }
    }

    /**
     * 当事件发生时,将事件加入队列
     *
     * @param event
     */
    public <V> CompletableFuture<ResultEvent<V>> asyncSendEvent(@NonNull final Event event) {
        CompletableFuture<ResultEvent<V>> completableFuture = new CompletableFuture<>();
        completableFuture.exceptionally(throwable -> {
            ResultEvent<V> result = new ResultEvent<>(event);
            result.setCode(0);
            result.setMessage(ThreadUtils.getStackMessage(throwable));
            return result;
        });

        return asyncSendEvent(new ObserverEvent(event, completableFuture));
    }

    /**
     * 当事件发生时,将事件加入队列
     *
     * @param observerEvent
     */
    public <V> CompletableFuture<ResultEvent<V>> asyncSendEvent(@NonNull final ObserverEvent observerEvent) {
        EventSourceItem eventSourceItem = getConsumerListenerMap().get(observerEvent.getEvent().getEventSourceName());
        if (eventSourceItem == null) {
            getEventSourceQueueMap().get(observerEvent.getEvent().getEventSourceName()).offer(observerEvent);
        } else {
            eventSourceItem.listConsumer()
                    .parallelStream()
                    .forEach(consumerItem -> consumerItem.getEventQueue().offer(observerEvent));
        }
        return observerEvent.getCompletableFuture();
    }


    public <V> ResultEvent<V> sendEvent(@NonNull final Event event) {
        try {
            CompletableFuture<ResultEvent<V>> completableFuture = asyncSendEvent(event);
            return completableFuture.get();
        } catch (Exception e) {
            ResultEvent<V> result = new ResultEvent<>(event);
            result.setCode(0);
            result.setMessage(ThreadUtils.getStackMessage(e));
            return result;
        }
    }

    public void sendAsyncEvent(@NonNull final Event event) {
        ObserverEvent observerEvent = new ObserverEvent(event, null);
        EventSourceItem eventSourceItem = getConsumerListenerMap().get(event.getEventSourceName());
        if (eventSourceItem == null) {
            getEventSourceQueueMap().get(event.getEventSourceName()).offer(observerEvent);
        } else {
            eventSourceItem.listConsumer()
                    .parallelStream()
                    .forEach(consumerItem -> consumerItem.getEventQueue().offer(observerEvent));
        }
    }

    public boolean hasEvent(@NonNull String eventSourceName, @NonNull String consumerName) {
        EventSourceItem eventSourceItem = getConsumerListenerMap().get(eventSourceName);
        if (eventSourceItem == null) {
            return Boolean.FALSE;
        } else {
            ConsumerItem consumerItem = eventSourceItem.getConsumer(consumerName);
            if (consumerItem == null) {
                return Boolean.FALSE;
            }
            return consumerItem.getEventQueue().peek() != null;
        }
    }

    /**
     * 从队列头开始，消费事件
     */
    public ObserverEvent getEvent(@NonNull String eventSourceName, @NonNull String consumerName) {
        EventSourceItem eventSourceItem = getConsumerListenerMap().get(eventSourceName);
        if (eventSourceItem == null) {
            return null;
        } else {
            ConsumerItem consumerItem = eventSourceItem.getConsumer(consumerName);
            if (consumerItem == null) {
                return null;
            }
            return consumerItem.getEventQueue().poll();
        }
    }

    /**
     * 从队列头开始，获取指定数量的事件
     *
     * @param eventSourceName
     * @param consumerName
     * @param batchSize
     * @return
     */
    public List<ObserverEvent> getEvents(@NonNull String eventSourceName, @NonNull String consumerName, int batchSize) {
        EventSourceItem eventSourceItem = getConsumerListenerMap().get(eventSourceName);
        if (eventSourceItem == null) {
            return null;
        } else {
            ConsumerItem consumerItem = eventSourceItem.getConsumer(consumerName);
            if (consumerItem == null) {
                return null;
            }
            List<ObserverEvent> eventList = new ArrayList<>(batchSize);
            ObserverEvent event = consumerItem.getEventQueue().poll();
            for (int idx = 0; idx < batchSize && event != null; idx++, event = consumerItem.getEventQueue().poll()) {
                eventList.add(event);
            }
            return eventList;
        }
    }

    @EqualsAndHashCode
    private static class EventSourceItem implements Serializable {

        @Getter
        private final String eventSourceName;

        private final Map<String, ConsumerItem> consumerMap = new ConcurrentHashMap(16);

        public EventSourceItem(String eventSourceName) {
            this.eventSourceName = eventSourceName;
        }

        /**
         * 添加消费者
         *
         * @param consumer
         */
        public synchronized void addConsumer(@NonNull ConsumerItem consumer) {
            this.consumerMap.put(consumer.getConsumerName(), consumer);
        }

        public ConsumerItem getConsumer(@NonNull String consumerName) {
            return this.consumerMap.get(consumerName);
        }

        public ConsumerItem removeConsumer(@NonNull String consumerName) {
            return this.consumerMap.remove(consumerName);
        }

        public Set<ConsumerItem> listConsumer() {
            return Collections.unmodifiableSet(new HashSet<>(this.consumerMap.values()));
        }
    }

    private static class ConsumerItem implements Serializable {

        @Getter
        private final String consumerName;

        @Getter
        private final String eventSourceName;

        @Getter
        private final Queue<ObserverEvent> eventQueue;

        private final List<ObserverEventListener> listeners = new ArrayList<>(10);

        public ConsumerItem(String eventSourceName, String consumerName) {
            this.consumerName = consumerName;
            this.eventSourceName = eventSourceName;
            this.eventQueue = new ConcurrentLinkedQueue<>();
        }

        /**
         * 给事件源注册监听器
         *
         * @param listener
         */
        public synchronized void addEventListener(@NonNull ObserverEventListener listener) {
            this.listeners.add(listener);
        }

        public List<ObserverEventListener> listEventListener() {
            return Collections.unmodifiableList(this.listeners);
        }
    }

}